package com.qny.ai.service.impl;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.qny.ai.service.StreamingTTSService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Service
@RequiredArgsConstructor
public class StreamingTTSServiceImpl implements StreamingTTSService {
    
    private final QiniuVoiceService qiniuVoiceService;
    private final ObjectMapper objectMapper = new ObjectMapper();
    
    // 存储每个会话的文本缓冲区
    private final Map<String, StringBuilder> sessionTextBuffers = new ConcurrentHashMap<>();
    // WebSocket发送同步锁
    private final Map<String, Object> sessionLocks = new ConcurrentHashMap<>();
    
    @Override
    public CompletableFuture<Void> processStreamingTTS(String text, String voiceType, String encoding, 
                                                      WebSocketSession session, String sessionId, String messageId) {
        return CompletableFuture.runAsync(() -> {
            try {
                log.info("[STREAMING_TTS] Processing streaming TTS for session: {}, text length: {}", sessionId, text.length());
                
                // 发送TTS开始信号
                sendTTSEvent(session, "tts_start", "开始语音合成...");
                
                // 将文本分块处理，实现流式TTS效果
                String[] sentences = text.split("[。！？.!?]");
                int delay = 0;
                
                for (int i = 0; i < sentences.length; i++) {
                    String sentence = sentences[i].trim();
                    if (sentence.isEmpty()) continue;
                    
                    // 为每个句子添加标点符号（除了最后一个）
                    if (i < sentences.length - 1) {
                        sentence += text.charAt(text.indexOf(sentence) + sentence.length());
                    }
                    
                    final String finalSentence = sentence;
                    final int currentDelay = delay;
                    
                    // 延迟发送每个句子，模拟流式效果
                    CompletableFuture.delayedExecutor(currentDelay, java.util.concurrent.TimeUnit.MILLISECONDS)
                        .execute(() -> {
                            try {
                                // 调用TTS服务合成单个句子
                                String base64Audio = qiniuVoiceService.ttsToBase64(finalSentence, voiceType, encoding);
                                
                                if (base64Audio != null && !base64Audio.isEmpty()) {
                                    // 发送TTS音频数据
                                    sendTTSEvent(session, "tts_blob", base64Audio, encoding, messageId);
                                    log.debug("[STREAMING_TTS] Sent TTS chunk for session: {}, sentence: {}", sessionId, finalSentence);
                                } else {
                                    log.warn("[STREAMING_TTS] TTS returned empty result for sentence: {}", finalSentence);
                                }
                            } catch (Exception e) {
                                log.error("[STREAMING_TTS] Failed to process TTS chunk: ", e);
                            }
                        });
                    
                    // 每个句子之间延迟500ms
                    delay += 500;
                }
                
                log.info("[STREAMING_TTS] Streaming TTS scheduled for session: {}, chunks: {}", sessionId, sentences.length);
                
            } catch (Exception e) {
                log.error("[STREAMING_TTS] Failed to process streaming TTS: ", e);
                sendTTSEvent(session, "tts_error", "语音合成时发生错误: " + e.getMessage());
            }
        });
    }
    
    @Override
    public void processTextChunk(String textChunk, String sessionId, boolean isFinal) {
        try {
            // 获取或创建会话的文本缓冲区
            StringBuilder buffer = sessionTextBuffers.computeIfAbsent(sessionId, k -> new StringBuilder());
            
            // 添加文本块到缓冲区
            buffer.append(textChunk);
            
            log.debug("[STREAMING_TTS] Processed text chunk for session: {}, size: {}, isFinal: {}", 
                     sessionId, textChunk.length(), isFinal);
            
            // 如果是最终块，处理完整的文本
            if (isFinal) {
                String fullText = buffer.toString();
                
                // 异步处理完整的文本
                CompletableFuture.runAsync(() -> {
                    try {
                        log.info("[STREAMING_TTS] Processing final text chunk for session: {}", sessionId);
                        
                        // 这里可以调用流式TTS处理
                        // 暂时使用现有的TTS服务
                        
                        // 清理缓冲区
                        sessionTextBuffers.remove(sessionId);
                        
                    } catch (Exception e) {
                        log.error("[STREAMING_TTS] Failed to process final text chunk: ", e);
                    }
                });
            }
            
        } catch (Exception e) {
            log.error("[STREAMING_TTS] Failed to process text chunk: ", e);
        }
    }
    
    @Override
    public void cleanupSession(String sessionId) {
        sessionTextBuffers.remove(sessionId);
        sessionLocks.remove(sessionId);
        log.info("[STREAMING_TTS] Cleaned up session: {}", sessionId);
    }
    
    private void sendTTSEvent(WebSocketSession session, String type, String content) {
        sendTTSEvent(session, type, content, null);
    }
    
    private void sendTTSEvent(WebSocketSession session, String type, String content, String mime) {
        sendTTSEvent(session, type, content, mime, null);
    }
    
    private void sendTTSEvent(WebSocketSession session, String type, String content, String mime, String messageId) {
        String sessionId = session.getId();
        Object lock = sessionLocks.computeIfAbsent(sessionId, k -> new Object());
        
        synchronized (lock) {
            try {
                // 检查WebSocket会话是否仍然打开
                if (!session.isOpen()) {
                    log.warn("[STREAMING_TTS] WebSocket session is closed, skipping event: {}", type);
                    return;
                }
                
                ObjectNode message = objectMapper.createObjectNode();
                message.put("type", type);
                
                // 对于tts_blob类型，使用payload字段；其他类型使用content字段
                if ("tts_blob".equals(type)) {
                    message.put("payload", content);
                } else {
                    message.put("content", content);
                }
                
                message.put("timestamp", System.currentTimeMillis());
                
                if (mime != null) {
                    message.put("mime", mime);
                }
                
                if (messageId != null && !messageId.isEmpty()) {
                    message.put("messageId", messageId);
                }
                
                session.sendMessage(new TextMessage(message.toString()));
                log.debug("[STREAMING_TTS] Sent event: {} - {}", type, content);
                
            } catch (Exception e) {
                log.error("[STREAMING_TTS] Failed to send TTS event: ", e);
            }
        }
    }
}
